package name.zxw.rabbitmq.basic;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author zhangxinwang
 * @date 2022/6/17 15:31
 */
public class Task2 {

    private static Integer COUNT_FOR = 5000;

    public static void main(String[] args) throws IOException, InterruptedException {
        long singleTime = singleConfirms();
        System.out.println("singleTime = " + singleTime);
        long unSingleTime = unSingleConfirms();
        System.out.println("unSingleTime = " + unSingleTime);
        long asyncTime = asyncConfirms();
        System.out.println("asyncTime = " + asyncTime);
    }

    public static Channel getChannel() {
        Channel channel = null;
        ConnectionFactory factory = new ConnectionFactory();
        try {
            channel = factory.newConnection().createChannel();
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (TimeoutException e) {
            throw new RuntimeException(e);
        }
        return channel;
    }

    public static long singleConfirms() throws IOException, InterruptedException {
        Channel channel = getChannel();
        channel.confirmSelect();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);
        long begin = System.currentTimeMillis();
        for (int i = 0; i < COUNT_FOR && channel.waitForConfirms(); i++) {
            channel.basicPublish("", queueName, null, new String("" + i).getBytes(StandardCharsets.UTF_8));
        }
        long end = System.currentTimeMillis();
        return end - begin;
    }

    public static long unSingleConfirms() throws IOException, InterruptedException {
        Channel channel = getChannel();
        channel.confirmSelect();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);
        long begin = System.currentTimeMillis();
        for (int i = 0; i < COUNT_FOR; i++) {
            channel.basicPublish("", queueName, null, new String("" + i).getBytes(StandardCharsets.UTF_8));
            if ((i + 1) % 1000 == 0 && channel.waitForConfirms()) {

            }
        }
        long end = System.currentTimeMillis();
        return end - begin;
    }

    public static long asyncConfirms() throws IOException, InterruptedException {
        Channel channel = getChannel();
        channel.confirmSelect();
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);

        ConcurrentSkipListMap<Long, String> map = new ConcurrentSkipListMap<>();
//        ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue<>();
        channel.addConfirmListener(
                (deliveryTag, multiple) -> {
                    if (multiple) {
                        map.headMap(deliveryTag).clear();
                    }
                    map.remove(deliveryTag);
                    System.out.println("success:" + deliveryTag + "---" + multiple);
                },
                (deliveryTag, multiple) -> {
                    System.out.println("error:" + deliveryTag + "---" + multiple);
                    map.get(deliveryTag);
                }
        );


        long begin = System.currentTimeMillis();
        for (int i = 0; i < COUNT_FOR; i++) {
            String message = "" + i;
            map.put(channel.getNextPublishSeqNo(), message);
//            queue.add(message);
            channel.basicPublish("", queueName, null, message.getBytes(StandardCharsets.UTF_8));
        }
        long end = System.currentTimeMillis();

        System.out.println("map = " + map);

        TimeUnit.SECONDS.sleep(5);
        System.out.println("map2 = " + map);

        return end - begin;
    }


}
